/*
此代码的功能是计算tf-idf
先看下原始数据
/data/input
这个目录下共508篇文章
[root@master input]# head 251it.seg.cln.txt
[| 液晶 电视 |] 产 自 中国 揭秘 朝鲜 小康 家庭 用 啥 电器 搜狐 数码 
第 页 [| 液晶 电视 |] 为 某 中国 品牌 
第 页 电冰箱 也 来自 中国 
第 页 主人 卧室 配备 上 了 [| 液晶 电脑 |] 
第 页 卧室 中 的 小 尺寸 电视机 
第 页 客厅 中 摆放 着 钢琴 
第 页 电热水器 
第 页 [| 抽 水 马桶 |] 
第 页 厨房 
第 页 一些 室内 元素 跟 中国 很 像 
[root@master input]# tail !$
tail 251it.seg.cln.txt
第 页 小区 楼房 
第 页 装饰 木门 
第 页 梳妆台 
第 页 挂历 
第 页 洗手盆 
第 页 厕所 
点击 图片 进入 下 一 页 电热水器 
上 一 页 上 一 页 
下 一 页 下 一 页 
分享 

虽说分好了词，但是分词并不是很好，bad case 没有在这里展示，而且好多标点符号没有过滤

1、将这508篇文章合并成一个文件(为了减少磁盘I/O，感觉这个有些牵强，一次性读取的时候也要磁盘I/O)，
而且一篇文章在新文件对应一行(方便计算tf，tf是以文章为准，所以可以在一行中求wordcount)
2、计算idf，存储在Map中，方便调用 idf = log(508/包含该词的文档数+1)
    a、计算wordBag。将所有文件的词语去重后得到一个词包
    b、针对所有的文章使用wordBag中的词去重
3、计算tf-idf = tf * idf
*/
package spark.example

import org.apache.spark._
import SparkContext._
import java.io.File

object tfIdf {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("compute tf-idf")
    val sc = new SparkContext(conf)
    val output = args(1)
    val dirPath = args(0).toString + "/" + "*.txt"
    import scala.util.matching.Regex
    val pattern = "[^\u4e00-\u9fa5]+" r
    val totalRDD = sc.textFile(dirPath)
    val wordTotalRDD = totalRDD.map(x => pattern.replaceAllIn(x, " ")).flatMap(x => x.trim().split(" "))
    val wordBagArray = wordTotalRDD.distinct().collect()
    import scala.math._
    // 这里使用wordBagArray 过滤，实际上是在做无用功，因为每一个文章中的单词都存在于wordBagArray 中，所有都符合。
    // 导致后面的idf 计算错误
    val idfMap = wordTotalRDD.filter(x => wordBagArray.contains(x))
      .map(x => (x, 1))
      .reduceByKey(_ + _)
      .map(x => (x._1, log(508.toDouble / (x._2 + 1).toDouble)))
      .collect()
      .toMap
    import scala.collection.mutable.ArrayBuffer
    val arraryBufferLine = new ArrayBuffer[String]()
    var index = 1001
    var lineString = ""
    for (fd <- subDir(new File("/data/input"))) {
      val lineArray = sc.textFile(fd.toString).map(x => pattern.replaceAllIn(x," ")).collect
      var lineString = ""
      for(line <- lineArray) {
        lineString = line + lineString
      }
      lineString = index.toString + "\t" + lineString
      index += 1
      arraryBufferLine += lineString
    }
    var tfArrayBuffer = new ArrayBuffer[Any]()
    //var tfIdfMap : Map[String,Any] = Map()
    for(line <- arraryBufferLine) {
      val wordCountFromLineRDD = sc.makeRDD(line.trim().split("\t")(1).split(" ")).filter(x => x.length >0).flatMap(x => x.split(" ")).map(x => (x,1)).reduceByKey(_+_)
      val maxWordCountFromLine = wordCountFromLineRDD.map(x => x._2).max
      val tfArray = wordCountFromLineRDD.map(x => (x._1,(x._2.toDouble / maxWordCountFromLine.toDouble) * idfMap(x._1))).collect()
      tfArrayBuffer += line.trim().split("\t")(0)
      tfArrayBuffer ++= tfArray
      //tfIdfMap += (line.trim().split("\t")(0).toString -> tfArray)
    }
    sc.makeRDD(tfArrayBuffer).saveAsTextFile(output)
  }

  def subDir(dir :File) :Iterator[File] = {
    val dirs = dir.listFiles().filter(_.isDirectory())
    val files = dir.listFiles().filter(_.isFile())
    files.toIterator ++ dirs.toIterator
  }

}